﻿using Confluent.Kafka;
using System;
using System.Collections.Generic;
using System.Text;

namespace Zhaoxi.Protobuf
{
	public class Consumer
	{
		public void Show()
		{
			string brokerList = "39.96.82.51:9093,47.95.2.2:9092,39.96.34.52:9092";
			string topicName = "test";
			var config = new ConsumerConfig
			{
				BootstrapServers = brokerList,
				GroupId = "mm",

				EnableAutoCommit = false,
				StatisticsIntervalMs = 5000,

				AutoOffsetReset = AutoOffsetReset.Earliest,
				EnablePartitionEof = true,

			};
			const int commitPeriod = 1;

			using (var consumer = new ConsumerBuilder<Ignore, NewInfo>(config)
				.SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))

				.SetPartitionsAssignedHandler((c, partitions) =>
				{
					Console.WriteLine($"Assigned partitions: [{string.Join(", ", partitions)}]");
				})
				.SetPartitionsRevokedHandler((c, partitions) =>
				{
					Console.WriteLine($"Revoking assignment: [{string.Join(", ", partitions)}]");
				})
					.SetValueDeserializer(new CustomStringIDeserializer<NewInfo>())
				.Build())
			{
				consumer.Subscribe(topicName);
				try
				{
					while (true)
					{
						try
						{
							var consumeResult = consumer.Consume();

							if (consumeResult.IsPartitionEOF)
							{
								//Console.WriteLine(
								//	$"Reached end of topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");

								continue;
							}

							Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Message.Value}");
							Console.WriteLine($": {consumeResult.TopicPartitionOffset}::{consumeResult.Message.Value.P_Title}：{consumeResult.Message.Value.P_Content}");

							if (consumeResult.Offset % commitPeriod == 0)
							{
								try
								{
									//consumer.Commit(consumeResult);
								}
								catch (KafkaException e)
								{
									Console.WriteLine($"Commit error: {e.Error.Reason}");
								}
							}
						}
						catch (ConsumeException e)
						{
							Console.WriteLine($"Consume error: {e.Error.Reason}");
						}
					}
				}
				catch (OperationCanceledException)
				{
					Console.WriteLine("Closing consumer.");
					consumer.Close();
				}
			}
		}
	}
}
